package cn.boat.router.producer;

// import com.oracle.jrockit.jfr.Producer;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;

public class ProducerLock {
    private String IP = "192.168.56.101:2181";

    private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<>();
    private static final ThreadLocal<ZooKeeper> ZK_THREAD_LOCAL = new ThreadLocal<>();

    private static final String Lock_ROOT_PATH = "/Locks";
    private static final String LOCK_NODE_NAME = "LOCK_";
    //ZooKeeper配置信息
    private ZooKeeper zkClient;
    //连接到Zookeeper服务器，同时将ZKClient自身作为默认的Watcher
    private String lockPath;



    // 监听lockPath的前一个节点的watcher
    private Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println(event.getPath() + " 前锁释放");
            System.out.println(event.getState() + " 前锁释放");
            synchronized (this) {
                notifyAll();
            }
        }
    };
    //每个客户端往持久化节点/parent_lock下创建有序临时节点/parent_lock/lock_。
    // 创建成功后/parent_lock下面会有每个客户端对应的节点，如/parent_lock/lock_000000001

    //打开zookeeper连接
    public ProducerLock() throws Exception {
        zkClient = new ZooKeeper("192.168.56.101:2181",
                5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
//                if (event.getType() == Event.EventType.None){
//                    if (
//                            event.getState() == Event.KeeperState.SyncConnected){
//                        System.out.println("连接成功");
//                    }
//                }
                if (event.getState() == Event.KeeperState.Disconnected) {
                    System.out.println("失去连接,连接消失");

                }

            }
        });
    }
    //获取锁
    public void getlock() throws Exception {
        //创建锁节点
        createLock();
        //尝试get锁
        attemptLock();
    }


    //创建锁，并创建线程节点
    private void createLock() throws Exception{
        //判断是否存在，如果节点不存在，则创建节点
        //位置，当前节点中数据，权限，持久化节点
        Stat stat = zkClient.exists(Lock_ROOT_PATH,false);
        if (stat == null){
            zkClient.create(Lock_ROOT_PATH,"node".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }
        //创建EPHEMERAL_SEQUENTIAL节点

        String lockPath = zkClient.create(Lock_ROOT_PATH + "/" + LOCK_NODE_NAME, new byte[0],
                ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("临时有序节点创建成功" );
        System.out.println(Thread.currentThread().getName() + " 锁创建: " + lockPath);
        this.lockPath = lockPath;

    }


    public void attemptLock() throws Exception{
        //获取Lock的所有节点，节点排序
        List<String> childlockPaths = null;
        childlockPaths = zkClient.getChildren(Lock_ROOT_PATH,false);
        Collections.sort(childlockPaths);

        for(String paths:childlockPaths){
            System.out.println("the uselockPath is: "+ paths );
        }

        System.out.println("当前最小节点是：" + lockPath.substring(Lock_ROOT_PATH.length() + 1));
        int index = childlockPaths.indexOf(lockPath.substring(Lock_ROOT_PATH.length() + 1));
        //若lockPath是序号最min的节点，则获取锁
        if (index == 0){
            System.out.println(Thread.currentThread().getName() + "获得锁*lockpath:"+lockPath);
            return;

        }
        else {
            //若不是最小节点，监控前一个
            String preLockPath = childlockPaths.get(index-1);
            Stat stat = zkClient.exists(Lock_ROOT_PATH + "/" + preLockPath, watcher);
            //继续若前一个也不存在了重新获取锁
            if (stat == null){
                attemptLock();
            }
            else{
                System.out.println("等待前方锁释放，preLockPath" + preLockPath);
                //同一时刻最多只有一个线程执行
                synchronized (watcher){
                    watcher.wait();

                }
                attemptLock();
            }
        }

    }
    //释放锁
    public void releaseLock() throws Exception {
        zkClient.delete(lockPath, -1);
        zkClient.close();
        System.out.println("锁释放:" + lockPath);
    }
    //测试
    public static void main(String[] args) throws Exception {
        try{
            ProducerLock producerLock = new ProducerLock();
            producerLock.createLock();
            producerLock.attemptLock();
            producerLock.releaseLock();
        }catch (Exception ex){
            ex.printStackTrace();
        }

    }
}
